package com.study.flink.java.day07_join_count;

import com.study.flink.java.day07_join_count.entity.ActBean;
import com.study.flink.java.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * 实时的分布式全局去重，可以使用redis效率较低，我们可以将用户的Id存储到State中
 * 1.定义一个state，state中存储的是hashSet，HashSet的特点是去重，但是HashSet的中的数据可能很大甚至内存溢出
 * 2.优化：定一个state，实用BloomFilter过滤器，利用Bitmap计算hash落位，缺点是不能计数，还要再定义一个state用来计数
 *
 */
public class ActivityCountV2 {

    public static void main(String[] args) throws Exception{
        ParameterTool parameters = ParameterTool.fromPropertiesFile(args[0]);
        DataStream<String> lines = FlinkUtils.createKafkaStream(parameters, SimpleStringSchema.class);

        //日志数据转换成Bean
        //u001,A1,2019-09-02 10:10:11,1,北京市
        SingleOutputStreamOperator<ActBean> actBeanStream = lines.map(new MapFunction<String, ActBean>() {
            @Override
            public ActBean map(String line) throws Exception {
                String[] fields = line.split(",");
                String uid = fields[0];
                String aid = fields[1];
                String time = fields[2];
                String date = time.split(" ")[0];
                int eventType = Integer.parseInt(fields[3]);
                String province = fields[4];
                return ActBean.of(uid, aid, date, eventType, province, 1);
            }
        });

        //按照指定的条件统计
        //统计次数
        //SingleOutputStreamOperator<ActBean> summed = actBeanStream.keyBy("aid", "date", "eventType").sum("count");
        //summed.print();

        //统计活动参与的人数【按照用户ID去重】
        KeyedStream<ActBean, Tuple> keyed = actBeanStream.keyBy("aid", "eventType");
        // 活动ID，事件类型，人数
        keyed.map(new RichMapFunction<ActBean, Tuple3<String, Integer, Long>>() {

            // 使用KeydState
            private transient ValueState<BloomFilter> uidState;
            private transient ValueState<Long> countState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //自定义一个布隆过滤器的状态描述器
                ValueStateDescriptor<BloomFilter> stateDescriptor = new ValueStateDescriptor<>(
                        "uid-state",
                        BloomFilter.class
                );

                //自定义一个计数器的状态描述器
                ValueStateDescriptor<Long> countDescriptor = new ValueStateDescriptor<>(
                        "count-state",
                        Long.class
                );

                //使用RuntimeContext获取状态
                uidState = getRuntimeContext().getState(stateDescriptor);
                countState = getRuntimeContext().getState(countDescriptor);
            }

            @Override
            public Tuple3<String, Integer, Long> map(ActBean actBean) throws Exception {
                String uid = actBean.uid;
                BloomFilter bloomFilter = uidState.value();
                Long count = countState.value();
                if (bloomFilter == null) {
                    //初始化一个bloomFilter
                    bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 1000000);
                    countState.update(0L);
                    count = 0L;
                }
                //BloomFilter可以判断一定不包含
                if (!bloomFilter.mightContain(uid)) {
                    // 将当前用户加入到bloomFilter
                    bloomFilter.put(uid);
                    count++;
                    countState.update(count);
                }
                //状态更新
                uidState.update(bloomFilter);
                return Tuple3.of(actBean.aid, actBean.eventType, count);
            }
        }).print();

        FlinkUtils.getEnv().execute("ActivityCountV2");

    }
    
}
